package com.winit.schedule.job.tracker.monitor;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.service.AbstractService;
import com.winit.schedule.job.tracker.JobTrackerContextImpl;

public class ZkHealthCheckMonitor extends AbstractService {

    // 用来定时发送心跳
    private final ScheduledExecutorService taskExecutor = Executors.newScheduledThreadPool(1);

    private JobTrackerContextImpl          jobTrackerContext;

    private ZKServiceMonitor               zkService;

    private static transient final Logger  logger       = LoggerFactory.getLogger(ZkHealthCheckMonitor.class);

    public ZkHealthCheckMonitor(String name, JobTrackerContextImpl jobTrackerContext){
        super(name);
        this.jobTrackerContext = jobTrackerContext;
    }

    public ZkHealthCheckMonitor(JobTrackerContextImpl jobTrackerContext){
        this("TaskTrackerMonitor", jobTrackerContext);
        zkService = jobTrackerContext.getZkService();
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {
        super.serviceInit(conf);
    }

    @Override
    protected void serviceStart() throws Exception {
        super.serviceStart();
        logger.info("--------> start zookeeper connection check");
        taskExecutor.scheduleWithFixedDelay(new CheckThread(), 5000, 1 * 60 * 1000, TimeUnit.MILLISECONDS);
    }

    @Override
    protected void serviceStop() throws Exception {
        super.serviceStop();
    }

    public JobTrackerContextImpl getJobTrackerContext() {
        return jobTrackerContext;
    }

    class CheckThread extends Thread {

        @Override
        public void run() {
            try {
                logger.info("--------> run jobtracker nodes check");
                ZooKeeper zk = zkService.getZooKeeper();

                List<String> childNodes = zk.getChildren(zkService.getPathServer(), false);
                if (null == childNodes || childNodes.size() == 0) {
                    logger.warn("----------> jobtracker register nodes is 0");
                }
            } catch (Throwable e) {
                logger.warn("-------- jobtracker nodes check------", e);
            }
        }

    }

}
